package com.self.raft;

import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

/**
 * 选举规则:
 *
 * 1. 在
 */
public class Election implements Runnable {

    /**
     * 响应队列(当前节点待发送给集群中其他参与者的投票队列)
     */
    private BlockingQueue<Vote> responseQueue = new LinkedBlockingQueue<>(20000);

    /**
     * 当前节点已收到的投票
     */
    private Set<Vote> receiveVoteSet = new HashSet<>();

    /**
     * 节点id到 Peer实例的映射
     */
    private Map<Integer,Peer> peerMap = new LinkedHashMap<>();

    /**
     * 集群节点(包含peer)
     */
    private List<Peer> allPeerList;

    /**
     * 当前节点
     */
    private Peer peer;

    /**
     * 当前选举周期
     */
    private int currentEpoch;

    /**
     * 当前节点所推举的leader节点
     */
    private int currentLeader;

    /**
     * 当前节点id
     */
    private int id;
    /**
     * 当前节点角色: 1:leader, 0: follower, -1: candidate
     */
    private int role;

    /**
     * 选举阶段: 1: 选举结束, 0:选举进行中
     */
    private int status = 0;

    private long heartDeadline;

    /**
     * 选举超时时间(毫秒)
     */
    private int electionTimeout = 150; // ms, random

    /**
     * 心跳超时时间(秒)
     */
    private int heartTimeout = 30; // unit: second

    /**
     * 当前节点是否存活
     */
    private AtomicBoolean alive = new AtomicBoolean(true);

    public Election(Peer peer, List<Peer> allPeer) {
        this.allPeerList = allPeer;
        this.peer = peer;
        this.role = 0;
        this.id = peer.id;
        this.currentEpoch = peer.currentEpoch;
        this.currentLeader = peer.currentLeader;
        for (Peer p : allPeerList) {
            peerMap.put(p.id,p);
        }

        // 启动Sender线程,负责把当前节点的投票发送给其他的节点(包含当前节点)
        new Sender().start();
    }

    public void start() {
        new Thread(this,"ElectionThread-"+peer.id).start();
    }

    @Override
    public void run() {

        while (alive.get()) {

            // 发送心跳,如果可以的话
            sendHeart();

            doElect();

            // follow 心跳检查
            checkHeart();

        }
    }

    private void sendVote(int state) {

        for (Peer peer : allPeerList) {
            Vote vote = new Vote(this.id,this.currentLeader,peer.id,this.currentEpoch,state);
            responseQueue.offer(vote);
        }
    }

    private void sendHeart() {

        // 只有leader节点才发送心跳
        while (status == 1 && this.role == 1) {
            sendVote(-1);
            Log.log(String.format("Leader: %s 发送心跳",this.currentLeader));
            // 在心跳超时时间内,要确保可以发送至少三次的心跳
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos((heartTimeout-1)/3));
        }
    }

    private void doElect() {

        Log.log("Start election processor");

        Random random = new Random();

        while (true) {

            // 选举超时随机时间[150,300) ms

            try {

                // 从当前节点的请求队列中获取投票请求
                Vote vote = this.peer.requestQueue.poll(electionTimeout+random.nextInt(electionTimeout), TimeUnit.MILLISECONDS);

                if(vote == null) {

                    int preEpoch = this.currentEpoch;

                    this.status = 0;
                    this.currentEpoch++; // 增加周期数
                    this.currentLeader = this.id; // 默认推举自己为leader

                    // 清空已收到的投票
                    receiveVoteSet.clear();

                    Log.log(String.format("超时未收到投票信息,开始新一轮投票,当前节点上个周期: %s, 新的投票周期: %s",preEpoch,this.currentEpoch));

                    if(this.role == 0) {
                        this.role = -1;
                        Log.log("当前节点由Follower >> 候选节点");
                    }

                    sendVote(1);

                } else {

                    // 心跳消息直接忽略
                    if(vote.getState() == -1) {
                        continue;
                    }

                    //Log.log(String.format("接收到投票: %s",vote));

                    if(vote.getState() == 0) {

                        /*
                         * 表明选举已结束,且收到的投票发送者为leader
                         */

                        int role = this.id == vote.getVotePeerId() ? 1 : 0;

                        this.currentEpoch = vote.getEpoch();
                        this.currentLeader = vote.getVotePeerId();
                        this.status = 1; // 选举结束
                        this.heartDeadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(heartTimeout);

                        Log.log(String.format("选举结束 >> 当前节点成为: %s, Leader: %s, 周期: %s",role == 1 ? "Leader" : "Follower",vote.getVotePeerId(),this.currentEpoch));

                        break; // 退出选举过程,转而进入心跳检测(follower角色)或发送心跳(leader)阶段

                    } else if(vote.getEpoch() > this.currentEpoch) {

                        /*
                         *
                         * 表明当前节点和其他节点已不再一个选举周期内,需要做:
                         * 1.使用当前投票更新当前节点的投票信息,
                         * 2.发送当前节点投票给其他节点
                         * 3.清空当前节点已收到的投票
                         */
                        this.currentEpoch = vote.getEpoch();
                        this.currentLeader = vote.getVotePeerId();

                        receiveVoteSet.clear(); //

                        // 告诉所有参与者,当前节点的投票信息
                        sendVote(1);

                    } else if(vote.getEpoch() < this.currentEpoch) {

                        // nothing to do. 收到的投票是过期投票,直接忽略.
                        // 这里未进行回复,该投票的发送者,超时未收到响应会自动提升选举周期

                    } else { // 选举周期一致


                        receiveVoteSet.add(vote);

                        // 是否要终止选举(this.currentLeader 的支持者过半)
                        if(shouldEnd()) {

                            this.role = this.currentLeader == this.id ? 1 : 0;
                            this.status = 1;

                            // 设置心跳有效期
                            this.heartDeadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(heartTimeout);

                            Log.log(String.format("选举结束 >>>> 当前节点成为: %s, Leader: %s, 周期: %s",role == 1 ? "Leader" : "Follower",this.currentLeader,this.currentEpoch));

                            if(this.currentLeader == this.id) {
                                // 当前节点为leader, 则通知集群内其他节点自己为leader
                                sendVote(0);
                            }

                            // 退出选举阶段
                            // 在这之前,无论是否接收到leader发来的vote.state == 0的投票, 节点都能够确定leader是谁
                            break;
                        }
                    }
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void checkHeart() {

        while (this.status == 1 && this.role != 1 && System.currentTimeMillis() < heartDeadline) {

            try {

                Vote vote = this.peer.requestQueue.poll(heartTimeout,TimeUnit.SECONDS);
                if(vote != null) {
                    Log.log(String.format("收到来自Leader: %s 的心跳",vote.getVotePeerId()));
                    this.heartDeadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(heartTimeout);
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        // 超时未收到Leader的心跳
        if(this.role != 1) {
            this.status = 0;
            // 将当前leader 删除
            Peer oldLeader = this.peerMap.remove(this.currentLeader);
            if(oldLeader != null) {
                this.allPeerList.remove(oldLeader);
            }

            Log.log(String.format("节点: %s, 未收到Leader: %s 的心跳, 开始新一轮选举",this.id,this.currentLeader));
        }
    }

    private boolean shouldEnd() {

        int num = 0;
        for (Vote vote : receiveVoteSet) {

            if(vote.getEpoch() == this.currentEpoch && vote.getVotePeerId() == this.currentLeader) {
                num++;
            }
        }

        return num > (allPeerList.size()/2 + 1);
    }

    private class Sender extends Thread {

        Sender() {
            setName("SenderThread-"+id);
            setDaemon(true);
        }

        @Override
        public void run() {
            super.run();

            while (true) {

                try {

                    Vote vote = responseQueue.poll(500,TimeUnit.MILLISECONDS);
                    if(vote != null) {
                        //Log.log(String.format("发送投票信息: %s", vote));

                        // 给投票数据中的接收者发送投票数据
                        Peer peer = peerMap.get(vote.getAcceptPeerId());
                        peer.receiveVote(vote);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                    break;
                }
            }
        }
    }
}
